package com.config;

import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.SocketAcceptor;
import io.rsocket.core.RSocketConnector;
import io.rsocket.plugins.RSocketInterceptor;
import org.reactivestreams.Publisher;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.codec.CharSequenceEncoder;
import org.springframework.core.codec.StringDecoder;
import org.springframework.messaging.rsocket.RSocketConnectorConfigurer;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.concurrent.atomic.AtomicInteger;

@Configuration
public class ConfigForServer {

    private static final MimeType FOO_MIME_TYPE = MimeTypeUtils.parseMimeType("messaging/x.foo");
    @Bean
    public RSocketStrategies rsocketStrategies() {
        return RSocketStrategies.builder()
                .metadataExtractorRegistry(registry ->
                        registry.metadataToExtract(FOO_MIME_TYPE, String.class, "foo"))
                .build();
    }

    private static  final  CountingInterceptor countingInterceptor = new CountingInterceptor();
    private static class CountingInterceptor implements RSocket, RSocketInterceptor {

        private RSocket delegate;

        private final AtomicInteger fireAndForgetCount = new AtomicInteger();

        private final AtomicInteger metadataPushCount = new AtomicInteger();


        public int getFireAndForgetCount() {
            return this.fireAndForgetCount.get();
        }

        public int getMetadataPushCount() {
            return this.metadataPushCount.get();
        }

        @Override
        public RSocket apply(RSocket rsocket) {
            Assert.isNull(this.delegate, "Unexpected RSocket connection");
            this.delegate = rsocket;
            return this;
        }

        @Override
        public Mono<Void> fireAndForget(io.rsocket.Payload payload) {
            return this.delegate.fireAndForget(payload)
                    .doOnSuccess(aVoid -> this.fireAndForgetCount.incrementAndGet());
        }

        @Override
        public Mono<Void> metadataPush(io.rsocket.Payload payload) {
            return this.delegate.metadataPush(payload)
                    .doOnSuccess(aVoid -> this.metadataPushCount.incrementAndGet());
        }

        @Override
        public Mono<io.rsocket.Payload> requestResponse(io.rsocket.Payload payload) {
            return this.delegate.requestResponse(payload);
        }

        @Override
        public Flux<Payload> requestStream(io.rsocket.Payload payload) {
            return this.delegate.requestStream(payload);
        }

        @Override
        public Flux<io.rsocket.Payload> requestChannel(Publisher<Payload> payloads) {
            return this.delegate.requestChannel(payloads);
        }
    }

    @Bean
    public RSocketConnectorConfigurer customRSocketConnectorConfigurer(){

        return (RSocketConnector connector)->{

     	    connector.interceptors(registry -> registry.forResponder(countingInterceptor));
        };
    }

}
